package com.dmall.scf.action.kanban

import com.dmall.scf.dto.{KanbanFieldValue, RegisteredMoney}
import com.dmall.scf.utils.{DateUtils, MySQLUtils}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * @descrption
 * scf-spark 注册资本分布
 * @author wangxuexing
 * @date 2020/1/10
 */
object RegMoneyDistributionAction extends KanbanAction{
  val CLASS_NAME = this.getClass.getSimpleName().filter(!_.equals('$'))

  val RANGE_50W = BigDecimal(500000)
  val RANGE_100W = BigDecimal(1000000)
  val RANGE_500W = BigDecimal(5000000)
  val RANGE_1000W = BigDecimal(10000000)

  /**
   * 处理
   *
   * @param spark
   * @return
   */
  override def action(spark: SparkSession, args: Array[String]): DataFrame = {
    import spark.implicits._
    if(args.length < 2){
      throw new Exception("请指定是当前年(值为1)还是去年(值为2)：1|2")
    }
    val lastDay = DateUtils.addSomeDays(-1)
    val (starDate, endDate, filedId) = args(1) match {
      case "1" => (DateUtils.formatNormal2ShortDateStr(DateUtils.getFirstDateOfCurrentYear),
        DateUtils.formatNormal2ShortDateStr(lastDay), 44)
      case "2" => (DateUtils.getLastYearFirstStr(DateUtils.YYYYMMDD),
        DateUtils.getLastYearLastStr(DateUtils.YYYYMMDD), 45)
      case _ =>  throw new Exception("请传入正确的参数：是当前年(值为1)还是去年(值为2)：1|2")
    }

    val sql = s"""SELECT
                      id,
                      IFNULL(registered_money, 0) registered_money
                    FROM
                      scfc_supplier_info
                    WHERE
                      `status` = 3
                    AND yn = 1"""
    val allDimension = MySQLUtils.getDFFromMysql(spark, sql)
    val beanList =  allDimension.map(x => RegisteredMoney(x.getLong(0), x.getDecimal(1)))
    //val filterList =  SparkUtils.dataFrame2Bean[RegisteredMoney](allDimension, classOf[RegisteredMoney])
   val supplierIds = spark.sql(
      s"""
                   SELECT DISTINCT(a.company_id) supplier_ids
                    FROM wumart2dmall.wm_ods_cx_supplier_card_info a
                    JOIN wumart2dmall.wm_ods_jrbl_loan_dkzhxx b ON a.card_code = b.gshkahao
                    WHERE a.audit_status = '2'
                      AND b.jiluztai = '0'""").collect().map(_.getLong(0))
    val filterList = beanList.filter(x => supplierIds.contains(x.supplierId))

    val range1 =  spark.sparkContext.collectionAccumulator[Int]
    val range2 =  spark.sparkContext.collectionAccumulator[Int]
    val range3 =  spark.sparkContext.collectionAccumulator[Int]
    val range4 =  spark.sparkContext.collectionAccumulator[Int]
    val range5 =  spark.sparkContext.collectionAccumulator[Int]
    filterList.foreach(x => {

      if(RANGE_50W.compare(x.registeredMoney) >= 0){
        range1.add(1)
      } else if (RANGE_50W.compare(x.registeredMoney) < 0 && RANGE_100W.compare(x.registeredMoney) >= 0){
        range1.add(1)
      } else if (RANGE_100W.compare(x.registeredMoney) < 0 && RANGE_500W.compare(x.registeredMoney) >= 0){
        range2.add(1)
      } else if (RANGE_500W.compare(x.registeredMoney) < 0 && RANGE_1000W.compare(x.registeredMoney) >= 0){
        range3.add(1)
      } else if (RANGE_1000W.compare(x.registeredMoney) < 0){
        range4.add(1)
      }
    })
    val resultList = List(("50万元以下", range1.value.size()), ("50-100万元", range2.value.size()),
                          ("100-500万元", range3.value.size()),("500-1000万元", range4.value.size()),
                          ("1000万元以上", range5.value.size())).map(x => {
      KanbanFieldValue(1, lastDay, filedId, x._1, x._2.toString, "")
    })

    getDataFrame(resultList, spark)
  }
}
